package io.mft.commons.db;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.mft.commons.function.response.kv.KVMap;
import io.mft.commons.multithreading.worker.Work;
import io.mft.commons.multithreading.worker.WorkManager;
import io.mft.commons.util.CommonUtil;
/**
 * 批量日志管理
 * @说明：TODO
 * @类型名称：BatchPushLogManager
 * @创建者: 胡常建  
 * @创建时间: 2019年5月10日 下午2:45:31
 * @修改者: 胡常建  
 * @修改时间: 2019年5月10日 下午2:45:31
 */
public abstract class BaseAsyncBatchDbExecutor <T>{
	/**
	 *   数据行Db保存成功监听
	 * @说明：TODO
	 * @类型名称：DbRowInstanceSavedListener
	 * @创建者: 胡常建  
	 * @创建时间: 2019年10月14日 下午5:02:30
	 * @修改者: 胡常建  
	 * @修改时间: 2019年10月14日 下午5:02:30
	 */
	public static interface DbRowInstanceSavedListener{
		/**
		 *   保存成功时调用，业务逻辑处理方需要快速结束，以防止异步线程耗尽导致阻塞
		 * @说明: TODO
		 * @方法名称: onSaved
		 * @参数 @param success
		 * @返回类型 void    
		 * @创建者: 胡常建  
		 * @创建时间: 2019年10月14日 下午5:03:41
		 * @修改者: 胡常建  
		 * @修改时间: 2019年10月14日 下午5:03:41
		 */
		void onSaved(boolean success);
	}
	public interface Initor <T,AB extends BaseAsyncBatchDbExecutor<T>>{
		void init(AB asyncBatchDbExecutor);
	}
	class DbRowInstanceSavedListenerWraper implements Work{
		T obj;
		boolean success=false;
		@Override
		public Object handle() {
			try {
				DbRowInstanceSavedListener listener=(DbRowInstanceSavedListener)obj;
				listener.onSaved(success);
			} catch (Exception e) {
				logger.error("",e);
			}
			return true;
		}
	}
	protected static final Logger logger = LoggerFactory.getLogger(BaseAsyncBatchDbExecutor.class);
	protected int batchSize=1000;
	protected String tbName;
	protected String dataSourceId;
	protected static  HashMap<String, Object> map=new HashMap<String, Object>();
	/**/
	WorkManager dbRowInstanceSavedListenerWorkManager=null;
	protected <AB extends BaseAsyncBatchDbExecutor<T>> BaseAsyncBatchDbExecutor(Class<?>  owner,String aliasId,String dataSourceId,String tbName,int batchSize,int batchWorkerNumber) {
		this.tbName=tbName;
		this.batchSize=batchSize;
		this.dataSourceId=dataSourceId;
		String key=BaseAsyncBatchDbExecutor.class.getName()+"-"+aliasId+"-"+owner.getName()+"-"+dataSourceId+"-"+tbName;
		WorkManager workManager=WorkManager.getManager(key, batchWorkerNumber, batchWorkerNumber);
		for (int i = 0; i < batchWorkerNumber; i++) {
			workManager.acceptWork(work);
		}
		String keyDbRowInstanceSavedListener="DbRowInstanceSavedListener-"+BaseAsyncBatchDbExecutor.class.getName()+"-"+aliasId+"-"+owner.getName()+"-"+dataSourceId+"-"+tbName;
		dbRowInstanceSavedListenerWorkManager=WorkManager.getManager(keyDbRowInstanceSavedListener, batchWorkerNumber, batchWorkerNumber);
	}
	private BlockingQueue<T> queue=new LinkedBlockingQueue<T>();
	public int number() {
		return queue.size();
	}
	public void add(T object) {
		try {
			queue.add(object);
		} catch (Exception e) {
			logger.error("",e);
		}
	}
	
	private Work work=new Work() {
		
		@Override
		public Object handle() {
			List<KVMap> parameters=new ArrayList<KVMap>();
			List<T> objects=new ArrayList<T>();
			while (true) {
				try {
					T object=queue.poll();
					if (object!=null) {
						KVMap data=new KVMap();
						if (object instanceof Map) {
							Map map=(Map)object;
							data.putAll(map);
						}else {
							data=CommonUtil.convertToKVMap(object);
						}
						parameters.add(DbUtil.formatToUnderline(data));
						objects.add(object);
					}
					if (parameters.size()==batchSize||object==null) {
						if (parameters.size()>0) {
							try {
								List<List<KVMap>> lists=splitGroupUniqueBatch(parameters);
								lists.forEach(list->{
									try {
										if (list.size()==0) {
											return;
										}
										int[] results=batch(dataSourceId, tbName, list);
										if (results==null) {
											return;
										}
										int i=0;
										for (int s : results) {
											try {
												T obj=objects.get(i);
												if (obj instanceof DbRowInstanceSavedListener) {
													DbRowInstanceSavedListenerWraper wraper=new DbRowInstanceSavedListenerWraper();
													wraper.obj=obj;
													wraper.success=true;
													dbRowInstanceSavedListenerWorkManager.acceptWork(wraper);
												}
											} catch (Exception e) {
												logger.error("",e);
											}
											i++;
										}
									} catch (Exception e2) {
										logger.error("",e2);
									}
								});

							} catch (Exception e) {
								//queue.addAll(objects);
								parameters.clear();
								objects.clear();
								logger.error("",e);
							}
						}
						parameters.clear();
						objects.clear();
					}
					if (object==null) {
						TimeUnit.SECONDS.sleep(3);
					}
				} catch (Exception e) {
					logger.error("数据源["+dataSourceId+"]-数据表["+tbName+"]异步批量操作异常",e);
				}
			}
		}
	};
	protected List<List<KVMap>> splitGroupUniqueBatch(List<KVMap> parameters) {
		List<List<KVMap>> lists=new ArrayList<>();
		lists.add(parameters);
		return lists;
	}
	protected abstract int[] batch(String datasourceId,String tbName,List<KVMap> parameters);
}
